package cn.getech.data.development.sink.kudu

import java.lang

import cn.getech.data.development.sink.jdbc.config.CustomTableConfig
import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.api.java.tuple
import org.apache.flink.api.java.typeutils.RowTypeInfo
import org.apache.flink.streaming.api.datastream.{DataStream, DataStreamSink}
import org.apache.flink.table.api.TableSchema
import org.apache.flink.table.sinks.{RetractStreamTableSink, TableSink}
import org.apache.flink.table.utils.TableConnectorUtils
import org.apache.flink.types.Row

/**
 * kudu通用Sink
 * @param conf 配置参数
 */
class FlinkStreamKuduRetractSink(conf: CustomTableConfig) extends RetractStreamTableSink[Row] {
  val schema: TableSchema = TableSchema.builder.fields(conf.param_field_names, conf.param_class_type2DataType).build

  override def getTableSchema: TableSchema = schema
  override def consumeDataStream(dataStream: DataStream[tuple.Tuple2[lang.Boolean, Row]]): DataStreamSink[_] = {
    val sink = new FlinkStreamKuduUpsertSink(conf)
//    dataStream.print("=FlinkStreamKuduRetractSink==========================")
    dataStream.addSink(sink)
      .setParallelism(dataStream.getParallelism)
      .name(TableConnectorUtils.generateRuntimeName(this.getClass, getFieldNames))
  }

  override def getRecordType: TypeInformation[Row] = {
    new RowTypeInfo(schema.getFieldTypes, schema.getFieldNames)
  }

  override def configure(strings: Array[String], typeInformations: Array[TypeInformation[_]]): TableSink[tuple.Tuple2[lang.Boolean, Row]] = ???
}
